package com.shujia.state

import org.apache.flink.api.common.functions.RuntimeContext
import org.apache.flink.api.common.state.{ValueState, ValueStateDescriptor}
import org.apache.flink.configuration.Configuration
import org.apache.flink.contrib.streaming.state.RocksDBStateBackend
import org.apache.flink.runtime.state.filesystem.FsStateBackend
import org.apache.flink.streaming.api.CheckpointingMode
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.environment.CheckpointConfig.ExternalizedCheckpointCleanup
import org.apache.flink.streaming.api.functions.KeyedProcessFunction
import org.apache.flink.util.Collector

object Demo03CheckPoint {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    // 每1000ms做一次checkpoint
    env.enableCheckpointing(1000)
    // 高级选项(可选)
    // 设置CheckPoint的模式为EXACTLY_ONCE精确一次/完全一次(默认)
    env.getCheckpointConfig.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE)
    // 设置两个CheckPoint任务之间的时间间隔
    env.getCheckpointConfig.setMinPauseBetweenCheckpoints(500)
    // 设置CheckPoint的超时时间
    env.getCheckpointConfig.setCheckpointTimeout(60000)
    // 设置同一时刻最多能有多少个CheckPoint任务
    env.getCheckpointConfig.setMaxConcurrentCheckpoints(1)
    // 设置在任务取消时不清理CheckPoint保存的状态
    env.getCheckpointConfig.enableExternalizedCheckpoints(ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION)
    // 设置CheckPoint目录 使用文件系统作为状态后端（保存状态的地方）
    env.setStateBackend(new FsStateBackend("hdfs://master:9000/flink/wc/checkpoint"))

    //    new RocksDBStateBackend("hdfs://master:9000/flink/wc/checkpoint", true)

    val linesDS: DataStream[String] = env.socketTextStream("master", 8888)

    /**
     * 统计每个单词的数量(需要保证每个单词的状态)
     */
    linesDS
      .flatMap(_.split(","))
      .map(word => (word, 1))
      .keyBy(_._1)
      .process(new KeyedProcessFunction[String, (String, Int), (String, Int)] {
        /**
         * 每个并行度中会执行一次，一般用于跟外部系统建立连接，或者做一些初始化的工作
         */
        var valueState: ValueState[Int] = _

        override def open(parameters: Configuration): Unit = {
          // 获取当前任务的运行环境
          val ctx: RuntimeContext = getRuntimeContext
          // 构建一个单值状态描述 传入要保存的状态的类型及名称
          val valueStateDesc = new ValueStateDescriptor[Int]("cnt", classOf[Int])
          // 获取一个单值状态 用于保存每个Key的状态 每个并行度中的每个Key都会有自己的一个状态
          valueState = ctx.getState(valueStateDesc)

        }

        override def close(): Unit = super.close()

        override def processElement(value: (String, Int), ctx: KeyedProcessFunction[String, (String, Int), (String, Int)]#Context, out: Collector[(String, Int)]): Unit = {
          val word: String = value._1
          // 获取最新的状态
          val cnt: Int = valueState.value()
          val newCnt: Int = cnt + 1
          // 用新的值更新状态
          valueState.update(newCnt)
          out.collect((value._1, newCnt))

        }
      })
      .print()

    env.execute()


  }

}
